package com.tmsb.sink;

import com.tmsb.pojo.FairShopOrderNetwork;
import com.tmsb.utils.ConfUtil;
import com.tmsb.utils.RedisCluster;
import com.tmsb.utils.Util;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.JedisCluster;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

/**
 * Description: 用tmsb_fair_shop的order_network来获取每日进货数指标的sink
 * Version:1.0.0
 *
 * @author wxw
 * @date 2020/7/27
 */

public class OrderNetworkSink extends RichSinkFunction<FairShopOrderNetwork> implements CheckpointedFunction {
    //TODO 总数量
    private ListState<Long> totalquantityState;
    private Long total_quantity;
    //TODO 订单id集合,用于判断是否重复算过
    private ListState<Set> idState;
    private Set idSet;
    private ListState<Long> dateState;
    private Long dateValue;
    private SimpleDateFormat sdf;
    private JedisCluster jedis;
    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        totalquantityState.clear();
        dateState.clear();
        idState.clear();
        totalquantityState.add(total_quantity);
        dateState.add(dateValue);
        idState.add(idSet);

    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        total_quantity = 0L;
        dateValue = 0L;
        idSet = new HashSet();
        jedis = RedisCluster.getInstance(ConfUtil.getRedisCluster()).getRedisClusterList().get(0);
        sdf = new SimpleDateFormat("yyyyMMdd");

        ListStateDescriptor<Set> idSetStateDes =
                new ListStateDescriptor<>("idSetState", TypeInformation.of(new TypeHint<Set>() {}));
        ListStateDescriptor<Long> totalquantityStateDes =
                new ListStateDescriptor<>("totalquantityState", TypeInformation.of(new TypeHint<Long>() {}));
        ListStateDescriptor<Long> dateStateDes =
                new ListStateDescriptor<>("on_date", TypeInformation.of(new TypeHint<Long>() {}));

        totalquantityState = context.getOperatorStateStore().getListState(totalquantityStateDes);
        idState = context.getOperatorStateStore().getListState(idSetStateDes);
        dateState = context.getOperatorStateStore().getListState(dateStateDes);

        if (context.isRestored()) {
            total_quantity = totalquantityState.get().iterator().next();
            dateValue = dateState.get().iterator().next();
            idSet = idState.get().iterator().next();
        }

    }
    @Override
    public void close() throws Exception {
        super.close();
        jedis.close();
    }
    @Override
    public void invoke(FairShopOrderNetwork value, Context context) throws Exception {
      //  long  end = Util.getZeroPointTimestamps(value.getUpdate_time()-28800000);
        long orderTime = value.getOrder_time()-28800000;
        //TODO 每天0时清空状态
        if(orderTime>=dateValue){
            total_quantity = 0L;
            idSet = new HashSet();
            totalquantityState.clear();
            idState.clear();
            dateValue = Util.getTomorrowZeroPointTimestamps();
        }else if(orderTime < Util.getZeroPointTimestamps()){
            return;
        }
        //TODO 其余的就是今天的订单，计算每日进货数
        long quantity ;
        long id = value.getId();
      /*  Iterator<Set> iterator = idState.get().iterator();
        while (iterator.hasNext()){
            idSet = iterator.next();
        }*/
        //TODO 已付款，待配送
        if(value.getOrder_status()==1){
            //TODO 如果订单状态为1的已经来过一次，不要重复计算
            if(idSet.contains(id)){
                return;
            }
            //TODO 新增订单
            quantity = value.getQuantity();
            idSet.add(id);
        }else if(value.getOrder_status()==-1){
            //TODO 订单状态为-1，即取消订单
            if(!idSet.contains(id)){
                return;// TODO 如果订单号不在set里面，就不管
            }
            quantity = value.getQuantity()*(-1);
            idSet.remove(id);
        }else {
            //TODO 订单状态为-3，即已退款
            if(!idSet.contains(id)){
                return;
            }
            quantity = value.getQuantity()*(-1);
            idSet.remove(id);
        }
        total_quantity += quantity;
        String date = sdf.format(new Date(orderTime));
        //TODO 向redis发送数据
        jedis.hset(ConfUtil.getOrderNetworkQuantityRedisKey(),date,total_quantity+"");
    }
}
